

#include "IOScheduler.h"
#include "Coroutine.h"
#include "Scheduler.h"
#include <cassert>
#include <cerrno>
#include <cstdlib>
#include <cstring>
#include <iostream>
#include <stdexcept>
#include <sys/epoll.h>
#include <unistd.h>
#include <fcntl.h>


static bool debug = false;

namespace zepco {

IOManager *IOManager::GetThis() {
    return dynamic_cast<IOManager *>(Scheduler::GetThis());
}

IOManager::FDContext::EventContext &IOManager::FDContext::getEventContext(Event event) {
    assert(event == READ || event == WRITE);
    switch (event) {
        case READ:
            return read;
        case WRITE:
            return write;
        default:
            break;
    }
    // 传入非法参数
    throw std::invalid_argument("Unsupported event type");
}

void IOManager::FDContext::resetEventContext(EventContext &eventctx) {
    eventctx.scheduler = nullptr;
    eventctx.zephyr.reset();
    eventctx.callback = nullptr;
}

void IOManager::FDContext::triggerEventContext(Event event) {
    assert(event & events);

    // 删除该事件 不在关注 注册的io事件为一次性
    // 如果想持续关注需要 每次触发事件后重新注册
    events = (Event)(events & ~event);

    EventContext &ctx = getEventContext(event);
    if (ctx.callback) {
        ctx.scheduler->scheduleLock(&ctx.callback);
    } else if (ctx.zephyr) {
        ctx.scheduler->scheduleLock(&ctx.zephyr);
    }

    resetEventContext(ctx);
    return ;
}

IOManager::IOManager(size_t threads, bool use_caller, const std::string &name)
    : Scheduler(threads, use_caller, name), TimerManager() {
    m_epfd = epoll_create(5000);
    assert(m_epfd > 0);

    int ret = pipe(m_tickleFDs);
    assert(!ret);

    epoll_event readpipe_event;
    readpipe_event.events = EPOLLIN | EPOLLET; // ?? TODO:
    readpipe_event.data.fd = m_tickleFDs[0];

    ret = fcntl(m_tickleFDs[0], F_SETFL, O_NONBLOCK);
    assert(!ret);

    ret = epoll_ctl(m_epfd, EPOLL_CTL_ADD, m_tickleFDs[0], &readpipe_event);
    assert(!ret);

    contextResize(32);

    start();
}

IOManager::~IOManager() {
    stop();
    close(m_epfd);
    close(m_tickleFDs[0]);
    close(m_tickleFDs[1]);
    for (auto &i : m_fdContexts) {
        if (i) {
            delete i;
        }
    }
}

void IOManager::contextResize(size_t size) {
    m_fdContexts.resize(size);
    for (size_t i = 0; i < m_fdContexts.size(); i++) {
        if (m_fdContexts[i] == nullptr) {
            m_fdContexts[i] = new FDContext();
            m_fdContexts[i]->fd = i;
        }
    }
}

int IOManager::addEvent(int fd, Event event, std::function<void()> callback) {
    FDContext *fd_ctx = nullptr;
    std::shared_lock<std::shared_mutex> read_lock(m_mtx);

    // 查找fd_ctx
    if ((int)m_fdContexts.size() > fd) {
        fd_ctx = m_fdContexts[fd];
        read_lock.unlock();
    } else {
        read_lock.unlock(); // 读锁解锁
        std::unique_lock<std::shared_mutex> write_lock(m_mtx); // 写锁上锁
        contextResize(fd * 1.5);
        fd_ctx = m_fdContexts[fd];
    }

    std::lock_guard<std::mutex> lock(fd_ctx->mutex);
    if (fd_ctx->events & event) // fd监听的事件中不存在这个事件
        return -1;

    // add new event
    int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_ADD;
    epoll_event epevent;
    epevent.events = EPOLLET | fd_ctx->events | event;
    epevent.data.ptr = fd_ctx;
    int ret = epoll_ctl(m_epfd, op, fd, &epevent);
    if (ret) {
        std::cerr << "IOManager::addEvent()::epoll_ctl failed: " << strerror(errno) << std::endl;
        return -1;
    }

    m_pendingEventCount ++;

    // update fdcontext
    fd_ctx->events = (Event)(fd_ctx->events | event);

    // update eventcontext
    FDContext::EventContext &event_ctx = fd_ctx->getEventContext(event);
    assert(!event_ctx.scheduler && !event_ctx.callback && !event_ctx.zephyr);
    event_ctx.scheduler = Scheduler::GetThis();

    // 如果提供回调保存回调 否则保存当前协程
    if (callback)
        event_ctx.callback.swap(callback);
    else {
        event_ctx.zephyr = Zephyr::GetThis();
        assert(event_ctx.zephyr->getState() == Zephyr::RUNNING);
    }
    return 0;
}

bool IOManager::delEvent(int fd, Event event) {
    FDContext *fd_ctx = nullptr;
    std::shared_lock<std::shared_mutex> read_lock(m_mtx);

    // 查找fd_ctx
    if ((int)m_fdContexts.size() > fd) {
        fd_ctx = m_fdContexts[fd];
        read_lock.unlock();
    } else {
        read_lock.unlock(); // 读锁解锁
        return false;
    }

    std::lock_guard<std::mutex> lock(fd_ctx->mutex);

    if (!(fd_ctx->events & event))
        return false;

    // delete event
    Event new_event = (Event)(fd_ctx->events & ~event);
    int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = EPOLLET | new_event;
    epevent.data.ptr = fd_ctx;
    int ret = epoll_ctl(m_epfd, op, fd, &epevent);

    if (ret) {
        std::cerr << "delEvent::epoll_ctl failed: " << strerror(errno) << std::endl;
        return false;
    }

    // 待处理事件减一
    m_pendingEventCount --;
    fd_ctx->events = new_event; // ?? TODO:

    // update event context
    FDContext::EventContext &event_ctx = fd_ctx->getEventContext(event);
    fd_ctx->resetEventContext(event_ctx);
    return true;
}

bool IOManager::cancelEvent(int fd, Event event) {
    FDContext *fd_ctx = nullptr;
    std::shared_lock<std::shared_mutex> read_lock;

    if ((int)m_fdContexts.size() > fd) {
        fd_ctx = m_fdContexts[fd];
        read_lock.unlock();
    } else {
        read_lock.unlock();
        return false;
    }

    std::lock_guard<std::mutex> lock(fd_ctx->mutex);
    if (fd_ctx->events & event)
        return false;

    Event new_event = (Event)(fd_ctx->events & ~event);
    int op = fd_ctx->events ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = EPOLLET | new_event;
    epevent.data.ptr = fd_ctx;
    int ret = epoll_ctl(m_epfd, op, fd, &epevent);

    if (ret) {
        std::cerr << "cancelEvent::epoll_ctl failed: " << strerror(errno) << std::endl;
        return false;
    }

    m_pendingEventCount --;

    // tigger
    fd_ctx->triggerEventContext(event); // 触发事件回调
    return true;
}

bool IOManager::cancelAll(int fd) {
    FDContext *fd_ctx = nullptr;
    std::shared_lock<std::shared_mutex> read_lock(m_mtx);

    if ((int)m_fdContexts.size() > fd) {
        fd_ctx = m_fdContexts[fd];
        read_lock.unlock();
    } else {
        read_lock.unlock();
        return false;
    }

    std::lock_guard<std::mutex> lock(fd_ctx->mutex);

    if (!fd_ctx->events)
        return false;

    int op = EPOLL_CTL_DEL;
    epoll_event epevent;
    epevent.events = NONE;
    epevent.data.ptr = fd_ctx;

    int ret = epoll_ctl(m_epfd, op, fd, &epevent);
    if (ret) {
        std::cerr << "cancelAll::epoll_ctl failed: " << strerror(errno) << std::endl;
        return false;
    }

    if (fd_ctx->events & WRITE) {
        fd_ctx->triggerEventContext(WRITE);
        m_pendingEventCount --;
    }

    if (fd_ctx->events & READ) {
        fd_ctx->triggerEventContext(READ);
        m_pendingEventCount --;
    }

    assert(fd_ctx->events == NONE);
    return true;
}

void IOManager::tickle() {
    if (!hasIdleThreads()) { // 判断是否有线程处于空闲状态
        return;
    }

    // 如果有空闲线程 向管道写入一个字符 T 向等待在m_tickleFDs[0]的线程发送通知
    int ret = write(m_tickleFDs[1], "T", 1);
    assert(ret == 1);
}

bool IOManager::stopping() {
    uint64_t timeout = getNextTimer();
    return timeout == ~0ull && m_pendingEventCount == 0 && Scheduler::stopping();
}

void IOManager::idle() {
    static const uint64_t MAX_EVENTS = 256;
    std::unique_ptr<epoll_event[]> events(new epoll_event[MAX_EVENTS]);

    while (true) {
        if (debug)
            std::cout << "IOManager::idle(), run in thread: " << Thread::GetThreadID() << std::endl;

        if (stopping()) {
            if (debug)
                std::cout << "name: " << getName() << " idle exits in thread: " << Thread::GetThreadID() << std::endl;
            break;
        }

        int nready = 0;
        while (true) {
            static const uint64_t MAX_TIMEOUT = 5000;
            uint64_t next_timeout = getNextTimer(); // 获取下一个超时的定时器
            next_timeout = std::min(next_timeout, MAX_TIMEOUT);

            nready = epoll_wait(m_epfd, events.get(),  MAX_EVENTS, next_timeout);

            // EINTR --> retry
            if (nready < 0 && errno == EINTR) {
                continue;
            } else {
                break;
            }
        }


        // collect all timers overdue
        // 处理定时器导致的epollwait触发
        std::vector<std::function<void()>> cbs;
        listExpiredCb(cbs);

        if (!cbs.empty()) {
            for (const auto &cb : cbs) {
                scheduleLock(cb);
            }
            cbs.clear();
        }

        // collect add events nready
        for (int i = 0; i < nready; i++) {
            epoll_event &event = events[i];

            if (event.data.fd == m_tickleFDs[0]) {
                uint8_t dummy[256];
                while (read(m_tickleFDs[0], dummy, sizeof(dummy)) > 0);
                continue;
            }

            FDContext *fd_ctx = (FDContext *)event.data.ptr;
            std::lock_guard<std::mutex> lock(fd_ctx->mutex);


            // 如果式错误或挂起 将其转换为读写事件
            if (event.events & (EPOLLERR | EPOLLHUP))
                event.events |= (EPOLLIN | EPOLLOUT) & fd_ctx->events;

            // epollevent --> selfevent
            int real_events = NONE;
            if (event.events & EPOLLIN)
                real_events |= READ;

            if (event.events & EPOLLOUT)
                real_events |= WRITE;

            if ((fd_ctx->events & real_events) == NONE)
                continue;

            // 将触发的事件real_events从fd_ctx的events中移除
            int left_events = (fd_ctx->events & ~real_events);

            int op = (left_events) ? EPOLL_CTL_MOD : EPOLL_CTL_DEL;
            event.events = EPOLLET | left_events;
            int ret = epoll_ctl(m_epfd, op, fd_ctx->fd, &event);
            if (ret) {
                std::cerr << "idle::epoll_ctl failed: " << strerror(errno) << std::endl;
                continue;
            }

            if (real_events & READ) {
                fd_ctx->triggerEventContext(READ);
                m_pendingEventCount --;
            }

            if (real_events & WRITE) {
                fd_ctx->triggerEventContext(WRITE);
                m_pendingEventCount --;
            }
        }

        Zephyr::GetThis()->yield();
    }
}

void IOManager::onTimerInsertAtFront() {
    tickle();
}

}